package com.rrd.dw.mr.gzip.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.RecordType;
import com.ucredit.babel.common.model.nfcs.NfcsData;
import com.ucredit.babel.common.model.nfcs.NfcsReport;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * 上海资信报告-主表
 */

public class ShangHaiNfcsReportMain extends ParseBase {
    private Text val = new Text();
    private Text key = new Text();


    @Override
    public boolean parseJob(Mapper.Context context, BaseModel<?> model, String successful) throws IOException, InterruptedException {
        NfcsReport obj = (NfcsReport) model;
        NfcsData data = obj.getData();

        JSONObject dataJson = data.getData() == null ? new JSONObject() : JSON.parseObject(data.getData());

        JSONObject creditReportHeader = dataJson.getJSONObject("creditReportHeader") == null ? new JSONObject() : dataJson.getJSONObject("creditReportHeader");
        JSONObject personalInformation = dataJson.getJSONObject("personalInformation") == null ? new JSONObject() : dataJson.getJSONObject("personalInformation");
        JSONObject loanTransactionInformation = dataJson.getJSONObject("loanTransactionInformation") == null ? new JSONObject() : dataJson.getJSONObject("loanTransactionInformation");


        StringBuilder sb = getBaseInfo(obj, successful);
        StringBuilder TxSb = new StringBuilder(500);

        sb.append(creditReportHeader.getString("reason") == null ? "" : creditReportHeader.getString("reason")).append(CRT001);
        sb.append(creditReportHeader.getString("reportNo") == null ? "" : creditReportHeader.getString("reportNo")).append(CRT001);
        sb.append(creditReportHeader.getString("reportTime") == null ? "" : creditReportHeader.getString("reportTime")).append(CRT001);

        sb.append(personalInformation.getString("name") == null ? "" : personalInformation.getString("name")).append(CRT001);
        sb.append(personalInformation.getString("certType") == null ? "" : personalInformation.getString("certType")).append(CRT001);
        sb.append(personalInformation.getString("certNo") == null ? "" : personalInformation.getString("certNo")).append(CRT001);
        sb.append(personalInformation.getString("gender") == null ? "" : personalInformation.getString("gender")).append(CRT001);
        sb.append(personalInformation.getString("dateOfBirth") == null ? "" : personalInformation.getString("dateOfBirth")).append(CRT001);

        JSONObject maritalStatus = personalInformation.getJSONObject("maritalStatus") == null ? new JSONObject() : personalInformation.getJSONObject("maritalStatus");
        JSONObject highestEducation = personalInformation.getJSONObject("highestEducation") == null ? new JSONObject() : personalInformation.getJSONObject("highestEducation");
        JSONObject title = personalInformation.getJSONObject("title") == null ? new JSONObject() : personalInformation.getJSONObject("title");
        JSONObject homePhone = personalInformation.getJSONObject("homePhone") == null ? new JSONObject() : personalInformation.getJSONObject("homePhone");
        JSONObject phoneNumber = personalInformation.getJSONObject("phoneNumber") == null ? new JSONObject() : personalInformation.getJSONObject("phoneNumber");
        JSONObject email = personalInformation.getJSONObject("email") == null ? new JSONObject() : personalInformation.getJSONObject("email");
        JSONObject spouseWorkUnit = personalInformation.getJSONObject("spouseWorkUnit") == null ? new JSONObject() : personalInformation.getJSONObject("spouseWorkUnit");
        JSONObject spousePhoneNumber = personalInformation.getJSONObject("spousePhoneNumber") == null ? new JSONObject() : personalInformation.getJSONObject("spousePhoneNumber");
        JSONObject informationSummary = loanTransactionInformation.getJSONObject("informationSummary") == null ? new JSONObject() : loanTransactionInformation.getJSONObject("informationSummary");

        sb.append(maritalStatus.getString("maritalDetail") == null ? "" : maritalStatus.getString("maritalDetail")).append(CRT001);
        sb.append(maritalStatus.getString("getDate") == null ? "" : maritalStatus.getString("getDate")).append(CRT001);

        sb.append(highestEducation.getString("educationalDetail") == null ? "" : highestEducation.getString("educationalDetail")).append(CRT001);
        sb.append(highestEducation.getString("getDate") == null ? "" : highestEducation.getString("getDate")).append(CRT001);

        sb.append(title.getString("titleDetail") == null ? "" : title.getString("titleDetail")).append(CRT001);
        sb.append(title.getString("getDate") == null ? "" : title.getString("getDate")).append(CRT001);

        sb.append(homePhone.getString("homePhoneDetail") == null ? "" : homePhone.getString("homePhoneDetail")).append(CRT001);
        sb.append(homePhone.getString("getDate") == null ? "" : homePhone.getString("getDate")).append(CRT001);

        sb.append(phoneNumber.getString("phoneNumberDetail") == null ? "" : phoneNumber.getString("phoneNumberDetail")).append(CRT001);
        sb.append(phoneNumber.getString("getDate") == null ? "" : phoneNumber.getString("getDate")).append(CRT001);

        sb.append(email.getString("emailDetail") == null ? "" : email.getString("emailDetail")).append(CRT001);
        sb.append(email.getString("getDate") == null ? "" : email.getString("getDate")).append(CRT001);

        sb.append(personalInformation.getString("spouseName") == null ? "" : personalInformation.getString("spouseName")).append(CRT001);
        sb.append(personalInformation.getString("spouseCertType") == null ? "" : personalInformation.getString("spouseCertType")).append(CRT001);
        sb.append(personalInformation.getString("spouseCertNo") == null ? "" : personalInformation.getString("spouseCertNo")).append(CRT001);
        sb.append(personalInformation.getString("spouseGender") == null ? "" : personalInformation.getString("spouseGender")).append(CRT001);
        sb.append(personalInformation.getString("spouseDateOfBirth") == null ? "" : personalInformation.getString("spouseDateOfBirth")).append(CRT001);

        sb.append(spouseWorkUnit.getString("spouseWorkUnitDetail") == null ? "" : spouseWorkUnit.getString("spouseWorkUnitDetail")).append(CRT001);
        sb.append(spouseWorkUnit.getString("getDate") == null ? "" : spouseWorkUnit.getString("getDate")).append(CRT001);

        sb.append(spousePhoneNumber.getString("spousePhoneNumberDetail") == null ? "" : spousePhoneNumber.getString("spousePhoneNumberDetail")).append(CRT001);
        sb.append(spousePhoneNumber.getString("getDate") == null ? "" : spousePhoneNumber.getString("getDate")).append(CRT001);

        sb.append(informationSummary.getString("loanNumber") == null ? "" : informationSummary.getString("loanNumber")).append(CRT001);
        sb.append(informationSummary.getString("firstLoanDay") == null ? "" : informationSummary.getString("firstLoanDay")).append(CRT001);
        sb.append(informationSummary.getString("maximumCreditLine") == null ? "" : informationSummary.getString("maximumCreditLine")).append(CRT001);
        sb.append(informationSummary.getString("loanCeiling") == null ? "" : informationSummary.getString("loanCeiling")).append(CRT001);
        sb.append(informationSummary.getString("loanBalance") == null ? "" : informationSummary.getString("loanBalance")).append(CRT001);
        sb.append(informationSummary.getString("agreedMonthlyRepayment") == null ? "" : informationSummary.getString("agreedMonthlyRepayment")).append(CRT001);
        sb.append(informationSummary.getString("currentOverdueAmount") == null ? "" : informationSummary.getString("currentOverdueAmount")).append(CRT001);
        sb.append(informationSummary.getString("maximumOverdueAmount") == null ? "" : informationSummary.getString("maximumOverdueAmount")).append(CRT001);
        sb.append(informationSummary.getString("maximumOverduePeriod") == null ? "" : informationSummary.getString("maximumOverduePeriod")).append(CRT001);

        key.set(getKey());
        val.set(sb.toString().replace("\n", " ").replace("\r", " "));
        context.write(key, val);
        return true;

    }

    @Override
    public String getTableName() {
        return "kafka_babel_nfcs_report_main_inc";
    }

    @Override
    public String getSubType() {
        return ".main";
    }

    @Override
    public String getType() {
        return RecordType.NFCS_REPORT.toString();
    }

    @Override
    public String getComment() {
        return "上海资信报告-主表";
    }
}
